package com.thinkingcoder.tcmq.client;

import java.util.concurrent.atomic.AtomicLong;

import com.thinkingcoder.tcmq.core.model.Message;
import com.thinkingcoder.tcmq.core.model.Stat;

import lombok.Getter;

/**
 * message consumer.
 */
public class TcConsumer<T> {

    private String id;

    TcBroker broker;

    @Getter
    private TcListener listener;

    private static final AtomicLong idGen = new AtomicLong(0);

    public TcConsumer(TcBroker broker) {
        this.broker = broker;
        this.id = "CID-" + idGen.getAndIncrement();
    }

    public void autoCreateTopic(String topic, boolean allowAutoCreate) {
        broker.autoCreateTopic(topic, id, allowAutoCreate);
    }

    public void subscribe(String topic) {
        broker.subscribe(topic, id);
    }

    public void unSubscribe(String topic) {
        broker.unSubscribe(topic, id);
    }

    public Message<T> receive(String topic) {
        return broker.receive(topic, id);
    }

    public boolean ack(String topic, int readOffset) {
        return broker.ack(topic, id, readOffset);
    }

    public boolean ack(String topic, Message<?> message) {
        int offset = Integer.parseInt(message.getHeaders().get("X-offset"));
        return ack(topic, offset);
    }

    public void listen(String topic, TcListener<T> listener) {
        this.listener = listener;
        broker.addConsumer(topic, this);
    }

    public Stat stat(String topic) {
        return broker.stat(topic, id);
    }
}
